package com.quanxiaoha.xiaohashu.count.biz.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Maps;
import com.quanxiaoha.framework.common.util.JsonUtils;
import com.quanxiaoha.xiaohashu.count.biz.constant.MQConstants;
import com.quanxiaoha.xiaohashu.count.biz.constant.RedisKeyConstants;
import com.quanxiaoha.xiaohashu.count.biz.enums.FollowUnfollowTypeEnum;
import com.quanxiaoha.xiaohashu.count.biz.model.dto.CountFollowUnfollowMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author: Jovan
 * @date: 2025/3/21
 * @description: 计数: 粉丝数
 **/
@Component
@RocketMQMessageListener(
        consumerGroup = "xiaohashu_group_" + MQConstants.TOPIC_COUNT_FANS, // Group 组
        topic = MQConstants.TOPIC_COUNT_FANS // 主题 Topic
)
@Slf4j
public class CountFansConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String,Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            // 缓存队列最大容量
            .bufferSize(50000)
            // 每批次最多聚合 1000 条
            .batchSize(1000)
            .linger(Duration.ofSeconds(1))
            // 多久聚合一次
            .setConsumerEx(this::consumeMessage)
            .build();

    @Override
    public void onMessage(String body) {
        // 往 bufferTrigger 添加元素
        bufferTrigger.enqueue(body);
        log.info("## 消费到了 MQ 【计数: 粉丝数】, {}...", body);
    }

    private void consumeMessage(List<String> bodies) {
        log.info("#==> 聚合消息，size：{}", bodies.size());
        log.info("#==> 聚合消息，{}", JsonUtils.toJsonString(bodies));

        // List<String> 转 List<CountFollowUnfollowMqDTO>+
        List<CountFollowUnfollowMqDTO> countFollowUnfollowMqDTOs = bodies.stream()
                .map(body -> JsonUtils.parseObject(body, CountFollowUnfollowMqDTO.class))
                .toList();
        // 按目标用户进行分组
        Map<Long,List<CountFollowUnfollowMqDTO>> groupMap = countFollowUnfollowMqDTOs.stream()
                .collect(Collectors.groupingBy(CountFollowUnfollowMqDTO::getTargetUserId));

        // 按组汇总数据，统计出最终的计数
        // key 为目标用户 ID，value 为最终操作的计数
        Map<Long,Integer> countMap = Maps.newHashMap();

        for(Map.Entry<Long,List<CountFollowUnfollowMqDTO>> entry : groupMap.entrySet()){
            List<CountFollowUnfollowMqDTO> list = entry.getValue();
            // 最终的计数值，默认为0
            int finalCount = 0;
            for(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO : list){
                // 获取当前操作的计数值
                int count = countFollowUnfollowMqDTO.getType();

                // 根据操作类型，获取对应枚举
                FollowUnfollowTypeEnum followUnfollowTypeEnum = FollowUnfollowTypeEnum.valueOf(count);

                // 若枚举为空，跳到下一次循环
                if(Objects.isNull(followUnfollowTypeEnum)) {continue;}

                switch (followUnfollowTypeEnum){
                    // 如果为关注操作，粉丝数 +1
                    case FOLLOW -> finalCount += 1;
                    // 如果为取关操作，粉丝数 -1
                    case UNFOLLOW -> finalCount -= 1;
                }
            }
            // 将最终的计数值，放入 countMap 中
            countMap.put(entry.getKey(),finalCount);
        }
        log.info("#==> 聚合后的计数数据，{}", JsonUtils.toJsonString(countMap));

        // 更新 Redis
        countMap.forEach((targetUserId,count) -> {
            // Redis Key
            String redisKey = RedisKeyConstants.buildCountUserKey(targetUserId);
            // 判断 Redis 中 Hash 是否存在
            boolean  isExist = redisTemplate.hasKey(redisKey);

            // 若存在才会更新
            // 因为缓存设有过期时间，考虑到过期后，缓存会被删除，所以需要判断一下，存在才回去更新，否则会报错
            if(isExist){
                // 更新 Redis
                redisTemplate.opsForHash().increment(redisKey,RedisKeyConstants.FIELD_FANS_TOTAL,count);
            }

            // TODO:发送 MQ，计数数据落库
            // 发送 MQ 通知计数服务：计数数据落库
            // 构建消息体 DTO
            Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countMap)).build();

            // 异步发送 MQ 消息，提升接口响应速度
            rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_FANS_2_DB,message,new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("==> 【计数服务：粉丝数入库】 MQ 发送成功，SendResult：{}", sendResult);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("==> 【计数服务：粉丝数入库】 MQ 发送异常：", throwable);
                }
            });
        });
    }
}
